package com.young.study.mq.kafka;


import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

import java.util.*;
import java.util.concurrent.ExecutionException;

public class TopicOperation {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        deleteTopic();
        getTopicList();
    }

    public static void createTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        // 三个参数，1、主题名称，2、分区个数 3、副本因子
        NewTopic newTopic = new NewTopic("test-topic-create1", 1, (short)1);
        List<NewTopic> newTopicList=new ArrayList<>();
        newTopicList.add(newTopic);
        // 异步创建
        CreateTopicsResult topics = adminClient.createTopics(newTopicList);
        // 阻塞等待异步创建完成，如果这里没有等待创建完成，立即调用获取topic 的方法的话，可能会发现没有新创建的这个topic
        topics.all().get();
    }

    public static void  getTopicList() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        ListTopicsResult listTopics = adminClient.listTopics();
        KafkaFuture<Map<String, TopicListing>> mapKafkaFuture = listTopics.namesToListings();
        // 这个map 的key是topic 名称，value 包含名称以及 internal
        Map<String, TopicListing> listingMap = mapKafkaFuture.get();
        for (String s : listingMap.keySet()) {
            System.out.println(s);
            System.out.println(listingMap.get(s));
        }
    }

    public static void deleteTopic() throws ExecutionException, InterruptedException {
        AdminClient adminClient = getAdminClient();
        // 异步删除
        DeleteTopicsResult deleteTopicsResult = adminClient.deleteTopics(Collections.singletonList("test-topic-create"));
        // 阻塞等待删除操作完成，如果没有阻塞而立即获取所有的topic 则可能发现这里这个被删除的topic 还存在，但是当我们再次删除的时候就会出现topic 不存在的错误
        deleteTopicsResult.all().get();
        System.out.println(deleteTopicsResult);
    }

    private static AdminClient getAdminClient() {
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "192.168.199.201:9092");
        AdminClient adminClient = AdminClient.create(properties);
        return adminClient;
    }


}
